package com.deep.flink;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author sw
 * @create 2023-06-30 11:26
 */
public class Test15 {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        //创建用于输出的表
        String createOutDDL = "CREATE TABLE t1(\n" +
                "  uuid VARCHAR(20) PRIMARY KEY NOT ENFORCED,\n" +
                "  name VARCHAR(10),\n" +
                "  age INT,\n" +
                "  ts TIMESTAMP(3),\n" +
                "  `partition` VARCHAR(20)\n" +
                ")\n" +
                "PARTITIONED BY (`partition`)\n" +
                "WITH (\n" +
                "  'connector' = 'hudi',\n" +
                "  'path' = 'file:///tmp/t2',\n" +
                "  'table.type' = 'MERGE_ON_READ'" +
                ") ";
        tableEnv.executeSql(createOutDDL);

//        tableEnv.executeSql("INSERT INTO t1 VALUES\n" +
//                "  ('id1','Danny',23,TIMESTAMP '1970-01-01 00:00:01','par1'),\n" +
//                "  ('id2','Stephen',33,TIMESTAMP '1970-01-01 00:00:02','par1'),\n" +
//                "  ('id3','Julian',53,TIMESTAMP '1970-01-01 00:00:03','par2'),\n" +
//                "  ('id4','Fabian',31,TIMESTAMP '1970-01-01 00:00:04','par2'),\n" +
//                "  ('id5','Sophia',18,TIMESTAMP '1970-01-01 00:00:05','par3'),\n" +
//                "  ('id6','Emma',20,TIMESTAMP '1970-01-01 00:00:06','par3'),\n" +
//                "  ('id7','Bob',44,TIMESTAMP '1970-01-01 00:00:07','par4'),\n" +
//                "  ('id8','Han',56,TIMESTAMP '1970-01-01 00:00:08','par4');");

        Table table = tableEnv.sqlQuery(" select * from t1  ");
        table.printSchema();
        table.execute().print();

        env.execute();

    }
}
